package top.cwk.flume.plugin.fireye;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.cwk.flume.plugin.bo.FireeyeLog;
import top.cwk.flume.plugin.util.HostUtils;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author chenwk <br>
 * @Description: FireEyeInterceptor 火眼日志采集需求 定制 <br>
 * @date 2020/3/27 3:20 下午  <br>
 * @Copyright tydic.com
 */
public class FireEyeInterceptor implements Interceptor {

    //打印日志，便于测试方法的执行顺序
    private static final Logger logger = LoggerFactory.getLogger(FireEyeInterceptor.class);

    private static String INFO_FLAG = " INFO";
    private static String WARN_FLAG = " WARN";
    private static String ERROR_FLAG = " ERROR";
    private static String DEBUG_FLAG = " DEBUG";

    //是否打印出处理后的日志
    private Boolean printLog;

    private FireeyeLog fireeyeLog;

    public FireEyeInterceptor(FireeyeLog log,
                              Boolean printLog) {
        this.fireeyeLog = log;
        this.printLog = printLog;
    }

    /****
     * @Description: 时间校验，成功则转换为Long 毫秒格式的时间戳
     * @Param: [timeStr]
     * @return: long
     * @date: 2020/6/23 2:37 下午
     * @author: kangkang
     */
    public static long isValidDate(String timeStr) {
        try {
            DateTimeFormatter ftf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            LocalDateTime parse = LocalDateTime.parse(timeStr, ftf);
            return LocalDateTime.from(parse).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        } catch (DateTimeParseException e) {
            logger.error("日期格式化失败:{}",timeStr);
            return 0L;
        }

    }

    private synchronized String parseLogBody(String body, String logLevelFlag) {
        String[] logBody = body.split(logLevelFlag);
        if (logBody.length > 1) {
            return logBody[1];
        }
        return "";
    }

    private FireeyeLog packageLogContext(String body) {
        FireeyeLog newLog = new FireeyeLog();
        body = body.trim();
        if (body.length() > 23) {
            try {
                //2020-03-27 16:14:05,903 、需要兼容格式 2020/03/27 16:14:05,903
                String logTime = body.substring(0, 19);
                logTime = logTime.replaceAll("/","-");
                long timestamp = isValidDate(logTime);
                if (timestamp == 0) {
                    //时间非法，忽略该行数据
                    logger.error("日志解析失败，开头时间格式错误: logTime={},logBody=[{}]", logTime, body);
                    return null;
                }
                //ip地址
                newLog.setHostIp(fireeyeLog.getHostIp());
                //日志时间
                newLog.setLogTime(timestamp);
                //日志级别
                String logLevel = "INFO";
                String logBody = "";
                if (body.contains(INFO_FLAG)) {
                    logLevel = INFO_FLAG.trim();
                    logBody = parseLogBody(body, INFO_FLAG);
                } else if (body.contains(WARN_FLAG)) {
                    logLevel = WARN_FLAG.trim();
                    logBody = parseLogBody(body, WARN_FLAG);
                } else if (body.contains(ERROR_FLAG)) {
                    logLevel = ERROR_FLAG.trim();
                    logBody = parseLogBody(body, ERROR_FLAG);
                } else if (body.contains(DEBUG_FLAG)) {
                    logLevel = DEBUG_FLAG.trim();
                    logBody = parseLogBody(body, DEBUG_FLAG);
                } else {
                    //没有日志级别，整个作为日志消息体
                    logLevel = INFO_FLAG.trim();
                    logBody = body;
                }
                newLog.setLevel(logLevel);
                //实例标示
                newLog.setAppName(fireeyeLog.getAppName());
                //线程名称
                newLog.setThreadId(fireeyeLog.getThreadId());
                newLog.setThreadName(fireeyeLog.getThreadName());
                //系统编码
                newLog.setSystemCode(fireeyeLog.getSystemCode());
                //获取java类 - 第一个() 内的可能为类名称
                Pattern pattern = Pattern.compile("(?<=\\()[^\\)]+");
                Matcher matcher = pattern.matcher(body);
                if ("none".equals(newLog.getLogCaller()) && matcher.find()) {
                    newLog.setLogCaller(matcher.group());
                } else {
                    newLog.setLogCaller(newLog.getLogCaller());
                }
                //日志内容
                newLog.setAppLog(logBody);
                return newLog;
            } catch (Exception e) {
                logger.error("解析日志异常:{}", e.getMessage());
            }
        }
        logger.warn("日志格式异常:{}", body);
        return null;
    }

    @Override
    public void initialize() {
        logger.info("初始化拦截器 FireEyeInterceptor : {} -- {}", fireeyeLog, printLog);
    }

    @Override
    public Event intercept(Event event) {
        if (event.getBody() != null) {
            Map<String, String> headers = event.getHeaders();
            logger.debug("处理拦截数据 -- header : {}", event);
            String reqBody = new String(event.getBody());
            logger.debug("处理拦截数据 -- body : {}", reqBody);
            FireeyeLog body = packageLogContext(reqBody);
            if (body != null) {
                event.setBody(("[" + JSONObject.toJSONString(body) + "]").getBytes());
                logger.debug("处理后的格式:{}", new String(event.getBody()));
                return event;
            }
        }
        logger.warn("body格式异常:{}", event);
        return null;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> result = new ArrayList<>();
        Event event;
        for (Event e : events) {
            event = intercept(e);
            if (event != null) {
                result.add(event);
            }
        }
        if (printLog) {
            int i = 1;
            for (Event e : result) {
                logger.info("{} 行数据处理结果: {}", i++, new String(e.getBody()));
            }
        }
        return result;
    }

    @Override
    public void close() {
        logger.info("拦截器关闭：FireEyeInterceptor close ");
    }

    /**
     * 通过该静态内部类来创建自定义对象供flume使用，实现Interceptor.Builder接口，并实现其抽象方法
     */
    public static class Builder implements Interceptor.Builder {

        private static String hostIp;
        private static String appName;
        private static String threadId;
        private static String threadName;
        private static String logCaller;
        //系统编码
        private static String systemCode;
        private static Boolean printLog;

        /**
         * 该方法主要用来返回创建的自定义类拦截器对象
         *
         * @return
         */
        @Override
        public Interceptor build() {
            FireeyeLog fireeyeLog = new FireeyeLog();
            fireeyeLog.setAppName(appName);
            fireeyeLog.setHostIp(hostIp);
            if(StringUtils.isNotBlank(threadId)){
                fireeyeLog.setThreadId(Integer.valueOf(threadId));
            }
            fireeyeLog.setThreadName(threadName);
            fireeyeLog.setLogCaller(logCaller);
            fireeyeLog.setSystemCode(systemCode);
            return new FireEyeInterceptor(fireeyeLog, printLog);
        }

        /**
         * 用来接收flume配置自定义拦截器参数
         *
         * @param context 通过该对象可以获取flume配置自定义拦截器的参数
         */
        @Override
        public void configure(Context context) {
            hostIp = context.getString("hostIp");
            if (hostIp == null || hostIp.equals("none")) {
                //获取本地ip地址
                hostIp = HostUtils.getHostIP();
            }
            appName = context.getString("appName");
            systemCode = context.getString("systemCode");
            threadId = context.getString("threadId");
            threadName = context.getString("threadName");
            logCaller = context.getString("logCaller");
            printLog = context.getBoolean("printLog");
            if (printLog == null) {
                printLog = false;
            }
        }

    }

}
